package yuan.serializer;

import io.protostuff.ProtostuffIOUtil;
import io.protostuff.Schema;
import io.protostuff.runtime.RuntimeSchema;
import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serializer;
import yuan.Company;

import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.util.Map;

public class CompanyDeserializer implements Deserializer<Company> {


    @Override
    public void configure(Map<String, ?> map, boolean b) {

    }

    @Override
    public Company deserialize(String s, byte[] bytes) {

        if (bytes == null) {
            return null;
        }
       /* if (bytes.length < 8) {
            throw new SerializationException("dsfdsf");
        }

        final ByteBuffer buffer = ByteBuffer.wrap(bytes);
        int nameLen,addressLen;
        String name,address;

        nameLen = buffer.getInt();

        final byte[] nameBuffer = new byte[nameLen];
        buffer.get(nameBuffer);

        addressLen = buffer.getInt();

        final byte[] addressBuffer = new byte[addressLen];
        buffer.get(addressBuffer);

        try {
            name = new String(nameBuffer, "UTF-8");
            address = new String(addressBuffer, "UTF-8");
        } catch (UnsupportedEncodingException e) {
            throw new SerializationException("sdlxell!");
        }

        return new Company(name, address);*/
        final Company cp = new Company();
        final Schema schema = (Schema)RuntimeSchema.getSchema(Company.class);
        ProtostuffIOUtil.mergeFrom(bytes, cp, schema);


        return cp;

    }

    @Override
    public void close() {

    }
}
